package com.ass.rocketmq.base;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import java.util.function.BiConsumer;
import java.util.function.BiFunction;


@Log4j2
@Component
//用到的设计模式 模板设计模式
//事务类型的生产者的监听
public abstract class AbstractTransactionListener<M> implements TransactionListener{



    //执行对应的处理业务
    protected abstract BiConsumer<M,String> doService();

    //事务逻辑的设定
    //不能仅通过rocketmq事务id存储,就认为事务成功了 还得需要message里面的事务id结合本地事务 最终 看事务是否成功了
    //业务存储完 接着存储事务部分 保证事务的一致性
    protected abstract  BiConsumer<M,String> doLocalTransaction();

    //本地事务的检查，检查本地事务是否成功了
    // 这里默认检查事务的逻辑 是去redis中获取rocketmq消息的事务id 建议结合本地事务逻辑自定义
    protected abstract BiFunction<M,String,Boolean> checkLocalTransaction();

    //获得泛型中的实际类型
    protected abstract TypeReference<M> getModel();



    /*
    *   事务消息的几种状态
    *   提交事务消息，消费者可以看到此消息COMMIT_MESSAGE,
        回滚事务消息，消费者不会看到此消息ROLLBACK_MESSAGE,
        事务未知状态，需要调用事务状态回查，确定此消息是提交还是回滚UNKNOW;
    * */
    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        log.info("开始执行本地事务....");
        LocalTransactionState state;
        try{
            String body = new String(message.getBody());
            //业务数据
            M m = JSON.parseObject(body, getModel());
            String transactionId = message.getTransactionId();
            //处理业务+事务部分
            //保证业务部分和事务部分事务一致 要不然业务没报错但是事务报错了 就会出问题 所以就默认都用数据库的事务
            BiConsumer<M,String> service = doService();
            service.accept(m,transactionId);
            /**故意停止线程2分 看看 broker是否回查producer
              结果：确实执行 broker回查producer
              Thread.sleep(2000*60*60);
            提交事务消息，消费者可以看到此消息*/
            state = LocalTransactionState.COMMIT_MESSAGE;
            log.info("本地事务已提交。{}",message.getTransactionId());
        }catch (Exception e){
            log.error("执行本地事务失败。{}",e);
            //回滚事务消息，消费者不会看到此消息
            state = LocalTransactionState.ROLLBACK_MESSAGE;
        }
        return state;
    }





    /*
    *   事务消息的几种状态
    *   提交事务消息，消费者可以看到此消息COMMIT_MESSAGE,
        回滚事务消息，消费者不会看到此消息ROLLBACK_MESSAGE,
        事务未知状态，需要调用事务状态回查，确定此消息是提交还是回滚UNKNOW;
    * */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        log.info("开始回查本地事务状态。{}",messageExt.getTransactionId());
        //可以在这里记录时间可以利用redis锁，
        // 一旦超过了多久则直接回滚消息LocalTransactionState.ROLLBACK_MESSAGE;（消费者看不到消息）
        LocalTransactionState state;
        String transactionId = messageExt.getTransactionId();
        String body = new String(messageExt.getBody());
        //业务数据
        M m = JSON.parseObject(body, getModel());
        BiFunction<M, String,Boolean> check = checkLocalTransaction();
        if (Boolean.TRUE.equals(check.apply(m,transactionId))){
            //提交事务消息，消费者可以看到此消息
            state = LocalTransactionState.COMMIT_MESSAGE;
        }else {
            //事务未知状态，需要调用事务状态回查，确定此消息是提交还是回滚
            state = LocalTransactionState.UNKNOW;
        }
        log.info("结束本地事务状态查询：{}",state);
        return state;
    }

}
